package gnss_plugs

import (
	"dgo/goutils/dnet"
	gnssutils "dgo/goutils/gnss"
	"dgo/goutils/utils"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type GnssCaster struct {
	NtripCltOnlineCnt int32
	NtripSvrOnlineCnt int32
	sendingWorker_n   int32
	sendingFlag       int32

	id           string
	isTerminated int
	listener     net.Listener
	processor    IGnssCasterProcessor
	//onlinelist   map[uint32]*GnssCasterConn

	Monitor_Sending  *utils.MonitorSpeedSizeRec
	Monitor_SendFail *utils.MonitorSpeedSizeRec

	block_size            uint16
	max_sending_size      int
	SendingWorker_WorkMax int

	memPool *utils.MemBlockPool

	onlinelist    utils.GSyncMapGroup // handle:GnssCasterConn
	lock          *sync.RWMutex
	addr          string
	SendQueueType int // 0: 不使用队列发送(直接发送), 1:之前的模式(每个连接使用一个纤程进行发送数据), 2:使用聚合队列模式
	LogTypeStr    string
	DebugMode     int

	NetInfo *dnet.DNetInfo
	OnErr   func(sender *GnssCaster, err error)

	OnCheckConnTimeOutFunc  func(sender *GnssCasterConn) bool
	OnNtripSvrShake         func(sender *GnssCasterConn, requestParser *gnssutils.NtripRequest) int
	OnNtripCltShake         func(sender *GnssCasterConn, requestParser *gnssutils.NtripRequest) int
	OnNtripCltData          func(sender *GnssCasterConn, buf []byte, l int, recvbytes int64, sendbytes int64) int
	OnNtripSvrData          func(sender *GnssCasterConn, buf []byte, l int)
	OnNtripConnDisconnected func(sender *GnssCasterConn)
}

func NewGnssCaster(processor IGnssCasterProcessor) *GnssCaster {
	rval := &GnssCaster{
		processor:     processor,
		lock:          new(sync.RWMutex),
		SendQueueType: 0,
		NetInfo:       &dnet.DNetInfo{},
		//	workers:      utils.NewMultiGroupTask(),
	}
	rval.Monitor_SendFail = utils.NewMonitorSpeedSizeRec()
	rval.Monitor_Sending = utils.NewMonitorSpeedSizeRec()
	return rval
}

func NewGnssCasterWithID(processor IGnssCasterProcessor, id string) *GnssCaster {
	rval := NewGnssCaster(processor)
	rval.id = id
	return rval
}

func (this *GnssCaster) QueueTypeStatus() string {
	if this.SendQueueType == 2 {
		return fmt.Sprintf("[%d]worker:%d/%d, %s", this.SendQueueType, this.sendingWorker_n, this.SendingWorker_WorkMax, this.Monitor_Sending.Info())
	}
	return ""
}

/**
 *
 */
func (this *GnssCaster) CheckKickOut() {
checkfor:
	for {
		if this.isTerminated == 1 {
			break checkfor
		}
		time.Sleep(time.Second * 20)

		lvCheckFunc := this.OnCheckConnTimeOutFunc

		this.RangeOnlineList(func(ctx *GnssCasterConn) bool {
			if lvCheckFunc != nil {
				if lvCheckFunc(ctx) {
					ctx.RequestDisconnect(fmt.Sprintf("[超时]请求离线, %s", ctx.StatusString()))
				}
			} else {
				ctx.checkKickOut(30)
			}
			return true
		})
	}
}

func (this *GnssCaster) SetAddress(addr string) bool {
	if this.addr != addr {
		this.addr = addr
		return true
	}
	return false
}

func (this *GnssCaster) SetID(val string) bool {
	if this.id != val {
		this.id = val
		return true
	}
	return false
}

func (this *GnssCaster) add2Online(ctx *GnssCasterConn) {
	_, loaded := this.onlinelist.LoadOrStoreFunc2(ctx.handle, func() interface{} {
		return ctx
	}, nil)
	if loaded { // 不会到这里, 如果到这里,应该有BUG
		this.onlinelist.Set(ctx.handle, ctx)
		utils.Log.WarnTypef(this.LogTypeStr, "[BUG]handle重入?")
	} else {
		this.NetInfo.DoConnect()
	}
	//this.lock.Lock()
	//defer this.lock.Unlock()
	//n := len(this.onlinelist)
	//this.onlinelist[ctx.handle] = ctx
	//if n != len(this.onlinelist) {
	//	this.NetInfo.DoConnect()
	//}
}

func (this *GnssCaster) RangeOnlineList(cb func(ctx *GnssCasterConn) bool) {
	this.onlinelist.Range(func(key, value interface{}) bool {
		return cb(value.(*GnssCasterConn))
	})
}

func (this *GnssCaster) RangeOnlineListSort(lessfunc func(ctx1, ctx2 *GnssCasterConn) bool, cb func(ctx *GnssCasterConn) bool) {
	lst := this.onlinelist.ListSorted(func(iItm, jItm interface{}) bool {
		return lessfunc(iItm.(*GnssCasterConn), jItm.(*GnssCasterConn))
	})
	for i := 0; i < len(lst); i++ {
		value := lst[i]
		if !cb(value.(*GnssCasterConn)) {
			break
		}
	}
}

func (this *GnssCaster) removeOnline(ctx *GnssCasterConn) {
	if this.onlinelist.RemoveEx(ctx.handle) {
		this.NetInfo.DoDisconnect()
	}
	//this.lock.Lock()
	//defer this.lock.Unlock()
	//n := len(this.onlinelist)
	//delete(this.onlinelist, ctx.handle)
	//n1 := len(this.onlinelist)
	//if n1 < n {
	//	this.NetInfo.DoDisconnect()
	//}

}

func (this *GnssCaster) OnlineNum() int {
	return this.onlinelist.Count()
}

func (this *GnssCaster) FindCasterConn(ctxHandle uint32) *GnssCasterConn {
	v := this.onlinelist.Get(ctxHandle, nil)
	if v == nil {
		return nil
	}
	return v.(*GnssCasterConn)
	//this.lock.RLock()
	//defer this.lock.RUnlock()
	//itm := this.onlinelist[ctxHandle]
	//if itm != nil {
	//	return itm
	//}
	//return nil
}

func (this *GnssCaster) RequestDisconnectAll(reason string) {
	this.RangeOnlineList(func(ctx *GnssCasterConn) bool {
		ctx.RequestDisconnect(reason)
		return true
	})
}

func (this *GnssCaster) Close() {
	this.isTerminated = 1
	if this.listener != nil {
		this.listener.Close()
		utils.Log.InfoTypef(this.LogTypeStr, "CASTER(%s)侦听(%s)关闭", this.id, this.addr)
	}
}

func (this *GnssCaster) innerSendingWork() {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	if !atomic.CompareAndSwapInt32(&this.sendingFlag, 0, 1) {
		return
	}
	defer func() {
		this.sendingFlag = 0
	}()
	var n int
	for {
		n = 0
		this.onlinelist.Range(func(key, value interface{}) bool {
			if this.isTerminated == 1 {
				return false
			}
			conn := value.(*GnssCasterConn)
			if conn.sendingBegin() {
				n := atomic.AddInt32(&this.sendingWorker_n, 1)
				if this.SendingWorker_WorkMax > 0 && n >= int32(this.SendingWorker_WorkMax) { // 超过最大线程数
					atomic.AddInt32(&this.sendingWorker_n, -1)
					conn.sendingEnd()
				} else {
					go conn.sendingOnce(func(sender *GnssCasterConn) {
						atomic.AddInt32(&this.sendingWorker_n, -1)
						sender.sendingEnd()
					})
					n++
				}
			}
			return this.isTerminated == 0
		})
		if n == 0 {
			time.Sleep(time.Millisecond * 100)
		}
	}

}

func (this *GnssCaster) Open() {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}
	this.isTerminated = 0
	netListen, err := net.Listen("tcp4", this.addr)
	if err != nil {
		if this.OnErr != nil {
			//utils.Log.Errf("CASTER(%s)侦听(%s)出现异常:%v", this.id, this.addr, err)
			this.OnErr(this, err)
		}
		return
	}
	this.listener = netListen

	defer netListen.Close()

	// 开启心跳
	utils.GoFunc2(this.CheckKickOut)
	if len(this.LogTypeStr) == 0 {
		utils.Log.Infof("CASTER(%s)侦听(%s)成功,等待建立连接", this.id, this.addr)
	} else {
		utils.Log.InfoTypef(this.LogTypeStr, "CASTER(%s)侦听(%s)成功,等待建立连接", this.id, this.addr)
	}

	if this.SendQueueType == 2 {
		if this.memPool == nil {
			this.memPool = utils.NewMemBlockPool(1024)
		}
		this.max_sending_size = 1024 * 100
		this.block_size = 1024
		this.memPool.ConfigMemBlockSize(this.block_size)
		go this.innerSendingWork()
	}

	for this.isTerminated == 0 {
		_conn, err := netListen.Accept()
		if err != nil {
			continue
		}
		dnet.SetKeepAlive(_conn, false)
		if this.processor != nil {
			if this.processor.OnNtripConnAccept(_conn) == gnssutils.BOOL_TRUE {
				ctx := newCasterConn(_conn, this)
				ctx.connecttime = time.Now()
				this.add2Online(ctx)
				utils.GoFunc2(ctx.handleRecv)
			} else {
				_conn.Close()
			}
		} else {
			ctx := newCasterConn(_conn, this)
			ctx.connecttime = time.Now()
			this.add2Online(ctx)
			utils.GoFunc2(ctx.handleRecv)
		}

	}
}
